package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
import java.time.ZoneId;

/**
 * Author: Felix
 * Date: 2022/5/11
 * Desc: 交易域支付成功事务事实表
 */
public class DwdTradePayDetailSuc {
    public static void main(String[] args) throws Exception {

        // TODO 1. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(60*15));

        // TODO 2. 检查点相关设置(略)

        // TODO 3. 读取 Kafka dwd_trade_order_detail 主题数据，封装为 Flink SQL 表
        tableEnv.executeSql("" +
                "create table dwd_trade_order_detail( " +
                "id string, " +
                "order_id string, " +
                "user_id string, " +
                "course_id string, " +
                "course_name string, " +
                "province_id string, " +
                "date_id string, " +
                "create_time string, " +
                "original_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "ts string, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_detail", "dwd_trade_pay_detail_suc"));




        // TODO 4. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tableEnv.executeSql("create table topic_db(" +
                "`database` String, " +
                "`table` String, " +
                "`type` String, " +
                "`data` map<String, String>, " +
                "`old` map<String, String>, " +
                "`proc_time` as PROCTIME(), " +
                "`ts` string " +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db", "dwd_trade_pay_detail_suc"));



        // TODO 5. 筛选支付成功数据
        Table paymentInfo = tableEnv.sqlQuery("select " +
                        "data['id'] id, " +
                        "data['order_id'] order_id, " +
                        "data['payment_type'] payment_type, " +
                        "data['callback_time'] callback_time, " +
                        "`proc_time`, " +
                        "ts " +
                        "from topic_db " +
                        "where `table` = 'payment_info' "
               +
            // "and `type` = 'update' " +
                "and data['payment_status']='1602'"
        );
        tableEnv.createTemporaryView("payment_info", paymentInfo);



        // TODO 7. 关联 3 张表获得支付成功宽表
        Table resultTable = tableEnv.sqlQuery("" +
                "select " +
                "od.id order_detail_id, " +
                "od.order_id, " +
                "od.user_id, " +
                "od.course_id, " +
                "od.course_name, " +
                "od.province_id, " +
                "pi.payment_type payment_type_code, " +
                "pi.callback_time, " +
                "od.original_amount, " +
                "od.coupon_reduce, " +
                "od.final_amount, " +
                "pi.ts, " +
                "od.row_op_ts row_op_ts " +
                "from dwd_trade_order_detail od  " +
                "join payment_info pi  " +
                "on pi.order_id = od.order_id ");
        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 8. 创建 Kafka dwd_trade_pay_detail 表
        tableEnv.executeSql("create table dwd_trade_pay_detail_suc( " +
                "order_detail_id string, " +
                "order_id string, " +
                "user_id string, " +
                "course_id string, " +
                "course_name string, " +
                "province_id string, " +
                "payment_type_code string, " +
                "callback_time string, " +
                "original_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "ts string, " +
                "row_op_ts timestamp_ltz(3), " +
                "primary key(order_detail_id) not enforced " +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_suc"));


       //tableEnv.executeSql("select * from dwd_trade_order_detail").print();
        //tableEnv.executeSql("select * from result_table").print();
        //tableEnv.executeSql("select * from topic_db").print();
        //tableEnv.executeSql("select * from payment_info").print();



        // TODO 9. 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("" +
                "insert into dwd_trade_pay_detail_suc select * from result_table");


    }


}
